﻿#include "AsyncLogger.h"

namespace AsyncLogger {

	AsyncLogger::AsyncLogger(const std::string& logDir, const std::string& basename, off_t rollSize, int flushInterval)
		:logDir_(logDir), basename_(basename), latch_(1), rollSize_(rollSize), flushInterval_(flushInterval), output_(logDir_, basename_, rollSize, false, flushInterval)
	{
		currBuffer_ = std::unique_ptr<Buffer>(new Buffer);
		nextBuffer_ = std::unique_ptr<Buffer>(new Buffer);
		isRunning_ = false;
		count_ = 0;
	}

	void AsyncLogger::start()
	{
		isRunning_ = true;
		logThread_ = std::move(std::unique_ptr<std::thread>(new std::thread(std::bind(&AsyncLogger::asyncLogThread, this))));
		latch_.wait();
	}

	void AsyncLogger::stop()
	{
		isRunning_ = false;
		condition_.notify_one();
		logThread_->join();
		if (currBuffer_) {
			output_.append(currBuffer_->data(), currBuffer_->length());
		}
		for (const auto& buffer : buffers_) {
			assert(buffer);
			output_.append(buffer->data(), buffer->length());
		}
		output_.flush();
	}

	AsyncLogger::~AsyncLogger()
	{
		if (isRunning_) {
			stop();
		}
	}

	void AsyncLogger::append(const char* logline, size_t len)
	{
		std::lock_guard<std::mutex> lock(mutex_);
		if (currBuffer_->avail() >= len) {
			currBuffer_->append(logline, len);
		}
		else {
			buffers_.emplace_back(std::move(currBuffer_));
			if (nextBuffer_) {
				currBuffer_ = std::move(nextBuffer_);
				nextBuffer_ = nullptr;
			}
			else {
				currBuffer_ = std::unique_ptr<Buffer>(new Buffer);
			}
			currBuffer_->append(logline, len);
		}
	}

	void AsyncLogger::asyncLogThread()
	{
		latch_.countDown();
		BufferPtr newBuffer1(new Buffer);
		BufferPtr newBuffer2(new Buffer);
		newBuffer1->bzero();
		newBuffer2->bzero();
		BufferVec buffersToWrite;
		//LogFile output(logDir_, basename_, 4096, false);
		std::cout << "start async log thread at: ";
		while (isRunning_) {
			assert(newBuffer1 != nullptr);
			assert(newBuffer2 != nullptr);
			assert(buffersToWrite.empty());
			// copy-on-write
			{
				std::unique_lock<std::mutex> lock(mutex_);
				condition_.wait_for(lock, std::chrono::seconds(3));
				buffers_.emplace_back(std::move(currBuffer_));
				currBuffer_ = std::move(newBuffer1);
				if (!nextBuffer_) {
					nextBuffer_ = std::move(newBuffer2);
				}
				buffersToWrite.swap(buffers_);
			}
			// 3s内出现大量日志，需要丢弃
			if (buffersToWrite.size() > 26) {
				char buf[256];
				snprintf(buf, sizeof(buf), "Dropped log messages at %s, %zd larger buffers\n",
					Timestamp::getFormatDate().c_str(), buffersToWrite.size() - 2);
				::fputs(buf, stderr);
				output_.append(buf, sizeof(buf));
				buffersToWrite.erase(buffersToWrite.begin() + 2, buffersToWrite.end());
			}
			// 处理日志
			for (const auto& buffer : buffersToWrite) {
				output_.append(buffer->data(), buffer->length());
			}
			if (buffersToWrite.size() > 2) {
				buffersToWrite.resize(2);
			}
			// 更新buffer1、buffer2
			if (!newBuffer1) {
				assert(!buffersToWrite.empty());
				newBuffer1 = std::move(buffersToWrite.back());
				buffersToWrite.pop_back();
				newBuffer1->reset();
			}
			if (!newBuffer2) {
				assert(!buffersToWrite.empty());
				newBuffer2 = std::move(buffersToWrite.back());
				buffersToWrite.pop_back();
				newBuffer2->reset();
			}
			buffersToWrite.clear();
			output_.flush();
		}
		output_.flush();
	}
}